from __future__ import annotations

"""
管理员广播下单路由（broadcast）
------------------------------
职责：
- 管理员对分组内多账户按账户余额与投入百分比广播下单
- 管理广播分组（增删改查、成员维护）
- 查看分组列表与风控配置

说明：
- 应用统一白名单与最小名义金额规则；对每个账户归一化数量/价格
"""

import logging
from decimal import Decimal
from typing import Any, Dict, List, Optional

from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
from sqlalchemy import delete, func, or_, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from ..config import get_settings
from ..db import get_session
from ..deps import require_admin, require_user
from ..models import (
    AuditLog,
    ExecOrder,
    Group,
    GroupMember,
    MarketType,
    OrderStatus,
    OrderType,
    PositionAction,
    TradeSide,
)
from ..schemas import (
    BroadcastOrderRequest,
    BroadcastResponse,
    BroadcastResultItem,
    GroupCreateRequest,
    GroupListResponse,
    GroupMemberCreateRequest,
    GroupMemberRead,
    GroupMemberUpdateRequest,
    GroupRead,
    GroupSummary,
    GroupUpdateRequest,
    GroupJoinRequest,
)
from ..services.hb_client import HBClientError, hb_client
from ..services.risk import RiskError, enforce_max_notional, enforce_whitelists, get_reference_price
from ..services.signal_worker import SignalWorker, signal_worker
from ..services.utils import default_account_name, translate_error_message

logger = logging.getLogger(__name__)

PLACEHOLDER_CONNECTOR = "default_connector"

router = APIRouter(prefix="/admin", tags=["broadcast"])
public_router = APIRouter(prefix="/broadcast", tags=["broadcast"])


async def _load_group_members(session: AsyncSession, group_ids: List[int]) -> Dict[int, List[GroupMemberRead]]:
    """一次性加载多个分组的成员列表。"""
    if not group_ids:
        return {}
    members_stmt = (
        select(GroupMember)
        .where(GroupMember.group_id.in_(group_ids))
        .order_by(GroupMember.group_id, GroupMember.id)
    )
    member_result = await session.execute(members_stmt)
    grouped: Dict[int, List[GroupMemberRead]] = {gid: [] for gid in group_ids}
    for member in member_result.scalars().all():
        grouped.setdefault(member.group_id, []).append(
            GroupMemberRead(
                account_name=member.account_name,
                connector_name=member.connector_name,
                weight=member.weight,
                enabled=member.enabled,
            )
        )
    return grouped


def _build_group_read(group: Group, members: List[GroupMemberRead]) -> GroupRead:
    return GroupRead(
        id=group.id,
        name=group.name,
        connector_name=group.connector_name,
        desc=group.desc,
        members=members,
    )


def _translate_error(code: Optional[str] = None, raw: Optional[str] = None) -> str:
    return translate_error_message(code=code, raw=raw)


def _extract_base_asset(symbol: str) -> str:
    if not symbol:
        return ""
    for separator in ("-", "_", "/", ":"):
        if separator in symbol:
            return symbol.split(separator)[0].upper()
    return symbol.upper()


def _extract_quote_asset(symbol: str) -> str:
    if not symbol:
        return ""
    for separator in ("-", "_", "/", ":"):
        if separator in symbol:
            return symbol.split(separator)[-1].upper()
    return symbol.upper()


def _safe_decimal(value: Any) -> Optional[Decimal]:
    if value is None:
        return None
    try:
        dec = Decimal(str(value))
    except Exception:  # noqa: BLE001
        return None
    if dec < 0:
        return Decimal("0")
    return dec


def _pick_balance_from_portfolio(portfolio: Any, quote_asset: str) -> Decimal:
    if not portfolio:
        return Decimal("0")
    quote_upper = (quote_asset or "").upper()

    def _match_asset(name: Any) -> bool:
        if not quote_upper:
            return False
        if not isinstance(name, str):
            return False
        return name.upper() == quote_upper

    if isinstance(portfolio, dict):
        balances = portfolio.get("balances")
        if isinstance(balances, dict):
            # 直接按键匹配
            for key, value in balances.items():
                if isinstance(key, str) and _match_asset(key):
                    if isinstance(value, dict):
                        for sub_key in ("available", "equity", "total", "balance", "free"):
                            dec = _safe_decimal(value.get(sub_key))
                            if dec is not None:
                                return dec
                    else:
                        dec = _safe_decimal(value)
                        if dec is not None:
                            return dec
            # 兼容 balances 内嵌对象（有 asset/currency 字段）
            for value in balances.values():
                if isinstance(value, dict):
                    asset_name = value.get("asset") or value.get("currency") or value.get("symbol") or value.get("name")
                    if _match_asset(asset_name):
                        for sub_key in ("available", "equity", "total", "balance", "free"):
                            dec = _safe_decimal(value.get(sub_key))
                            if dec is not None:
                                return dec
            # 若找不到特定资产，尝试读取聚合字段
            for key in ("equity", "total_equity", "available"):
                dec = _safe_decimal(balances.get(key))
                if dec is not None:
                    return dec
        if isinstance(balances, list):
            for item in balances:
                if not isinstance(item, dict):
                    continue
                asset_name = item.get("asset") or item.get("currency") or item.get("symbol") or item.get("name")
                if _match_asset(asset_name):
                    for sub_key in ("available", "equity", "total", "balance", "free"):
                        dec = _safe_decimal(item.get(sub_key))
                        if dec is not None:
                            return dec

        # 尝试其它聚合键
        for key in (
            f"{quote_upper}_equity" if quote_upper else None,
            f"{quote_upper}_balance" if quote_upper else None,
            "quote_equity",
        ):
            if key:
                dec = _safe_decimal(portfolio.get(key))
                if dec is not None:
                    return dec
        for key in ("total_equity", "portfolio_value", "equity", "net_value"):
            dec = _safe_decimal(portfolio.get(key))
            if dec is not None:
                return dec

    if isinstance(portfolio, list):
        for item in portfolio:
            if not isinstance(item, dict):
                continue
            asset_name = (
                item.get("token")
                or item.get("asset")
                or item.get("currency")
                or item.get("symbol")
                or item.get("name")
            )
            if _match_asset(asset_name):
                for key in ("available_units", "available", "equity", "total", "balance", "free", "units"):
                    dec = _safe_decimal(item.get(key))
                    if dec is not None:
                        return dec
        # 若列表未匹配具体资产，尝试聚合字段
        for aggregate_key in ("total_equity", "portfolio_value", "equity"):
            for item in portfolio:
                if isinstance(item, dict):
                    dec = _safe_decimal(item.get(aggregate_key))
                    if dec is not None:
                        return dec

    # 未找到则返回 0
    return Decimal("0")


async def _fetch_member_portfolio(account_name: str, connector: str) -> Any:
    try:
        portfolio = await hb_client.get_connector_portfolio_state(account_name, connector)
        if not portfolio:
            portfolio = await hb_client.get_portfolio_state(account_name)
    except HBClientError as exc:
        raise HBClientError(f"无法获取账户 {account_name} 的资产：{exc}") from exc
    return portfolio


def _infer_market_type(connector: str) -> MarketType:
    name = (connector or "").lower()
    if any(keyword in name for keyword in ("perp", "swap", "future", "futures")):
        return MarketType.perpetual
    return MarketType.spot


@router.get(
    "/groups",
    response_model=GroupListResponse,
    summary="查询广播分组列表",
    description=(
        "分页返回广播分组列表，可按名称或描述关键字过滤。响应包含总数、页码及每个分组的成员信息。"
    ),
)
async def list_groups(
    _: object = Depends(require_admin),
    session: AsyncSession = Depends(get_session),
    page: int = Query(1, ge=1, description="页码，从 1 开始"),
    page_size: int = Query(20, ge=1, le=100, description="每页条数（默认 20，上限 100）"),
    keyword: Optional[str] = Query(None, description="名称或描述关键词过滤"),
) -> GroupListResponse:
    filters = []
    if keyword:
        pattern = f"%{keyword.strip()}%"
        filters.append(or_(Group.name.ilike(pattern), Group.desc.ilike(pattern)))

    count_stmt = select(func.count()).select_from(Group)
    if filters:
        count_stmt = count_stmt.where(*filters)
    total_result = await session.execute(count_stmt)
    total = total_result.scalar_one()

    if total == 0:
        return GroupListResponse(total=0, page=page, page_size=page_size, items=[])

    stmt = select(Group)
    if filters:
        stmt = stmt.where(*filters)
    stmt = stmt.order_by(Group.id.desc()).offset((page - 1) * page_size).limit(page_size)
    result = await session.execute(stmt)
    groups = result.scalars().all()

    group_ids = [group.id for group in groups]
    members_map = await _load_group_members(session, group_ids)
    items = [_build_group_read(group, members_map.get(group.id, [])) for group in groups]
    return GroupListResponse(total=total, page=page, page_size=page_size, items=items)


@router.post(
    "/groups",
    response_model=GroupRead,
    status_code=status.HTTP_201_CREATED,
    summary="创建广播分组",
    description="管理员创建新的广播分组，可选填描述。创建后默认无成员。",
)
async def create_group(
    payload: GroupCreateRequest,
    admin=Depends(require_admin),
    session: AsyncSession = Depends(get_session),
) -> GroupRead:
    connector = payload.connector_name.strip()
    if not connector:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="connector_name 不能为空")
    group = Group(name=payload.name, connector_name=connector, desc=payload.desc)
    session.add(group)
    try:
        await session.commit()
    except IntegrityError as exc:
        await session.rollback()
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="分组名称已存在") from exc
    await session.refresh(group)
    session.add(
        AuditLog(
            actor_id=admin.id,
            action="broadcast.group.create",
            target=str(group.id),
            payload_json={"name": payload.name, "desc": payload.desc, "connector_name": connector},
        )
    )
    await session.commit()
    return _build_group_read(group, [])


@router.get(
    "/groups/{group_id}",
    response_model=GroupRead,
    summary="查看分组详情",
    description="返回指定广播分组的基础信息与成员列表。",
)
async def get_group(
    group_id: int,
    _: object = Depends(require_admin),
    session: AsyncSession = Depends(get_session),
) -> GroupRead:
    group = await session.get(Group, group_id)
    if not group:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="分组不存在")
    members_map = await _load_group_members(session, [group.id])
    return _build_group_read(group, members_map.get(group.id, []))


@router.patch(
    "/groups/{group_id}",
    response_model=GroupRead,
    summary="更新分组信息",
    description="支持修改广播分组的名称与描述。",
)
async def update_group(
    group_id: int,
    payload: GroupUpdateRequest,
    admin=Depends(require_admin),
    session: AsyncSession = Depends(get_session),
) -> GroupRead:
    group = await session.get(Group, group_id)
    if not group:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="分组不存在")
    if payload.name is not None:
        group.name = payload.name
    if payload.connector_name is not None:
        new_connector = payload.connector_name.strip()
        if not new_connector:
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="connector_name 不能为空")
        members_stmt = select(GroupMember).where(GroupMember.group_id == group_id)
        result = await session.execute(members_stmt)
        member_rows: List[GroupMember] = result.scalars().all()
        mismatched = [m for m in member_rows if m.connector_name and m.connector_name.lower() not in {PLACEHOLDER_CONNECTOR, new_connector.lower()}]
        if mismatched:
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="存在成员绑定的交易所与新设置不一致")
        group.connector_name = new_connector
        # 将占位符更新为新的交易所
        for member in member_rows:
            if member.connector_name.lower() == PLACEHOLDER_CONNECTOR:
                member.connector_name = new_connector
    if payload.desc is not None:
        group.desc = payload.desc
    try:
        await session.commit()
    except IntegrityError as exc:
        await session.rollback()
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="分组名称已存在") from exc
    await session.refresh(group)
    session.add(
        AuditLog(
            actor_id=admin.id,
            action="broadcast.group.update",
            target=str(group.id),
            payload_json=payload.dict(exclude_unset=True),
        )
    )
    await session.commit()
    members_map = await _load_group_members(session, [group.id])
    return _build_group_read(group, members_map.get(group.id, []))


@router.delete(
    "/groups/{group_id}",
    status_code=status.HTTP_204_NO_CONTENT,
    summary="删除分组",
    description="删除广播分组及其所有成员记录。",
)
async def delete_group(
    group_id: int,
    admin=Depends(require_admin),
    session: AsyncSession = Depends(get_session),
) -> Response:
    group = await session.get(Group, group_id)
    if not group:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="分组不存在")
    await session.execute(delete(GroupMember).where(GroupMember.group_id == group_id))
    await session.delete(group)
    session.add(
        AuditLog(
            actor_id=admin.id,
            action="broadcast.group.delete",
            target=str(group_id),
            payload_json={},
        )
    )
    await session.commit()
    return Response(status_code=status.HTTP_204_NO_CONTENT)


@router.post(
    "/groups/{group_id}/members",
    response_model=GroupMemberRead,
    status_code=status.HTTP_201_CREATED,
    summary="新增分组成员",
    description="为广播分组新增成员，需指定账户名，可选权重（暂不影响金额分配），重名将提示已存在。",
)
async def add_group_member(
    group_id: int,
    payload: GroupMemberCreateRequest,
    admin=Depends(require_admin),
    session: AsyncSession = Depends(get_session),
) -> GroupMemberRead:
    group = await session.get(Group, group_id)
    if not group:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="分组不存在")
    connector = payload.connector_name.strip()
    if connector.lower() != group.connector_name.lower():
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="成员交易所需与分组一致")
    if payload.weight < 1:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="权重必须为正整数")
    member = GroupMember(
        group_id=group_id,
        account_name=payload.account_name,
        connector_name=connector,
        weight=payload.weight,
        enabled=payload.enabled,
    )
    session.add(member)
    try:
        await session.commit()
    except IntegrityError as exc:
        await session.rollback()
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="该账户已在分组中") from exc
    await session.refresh(member)
    session.add(
        AuditLog(
            actor_id=admin.id,
            action="broadcast.group.member.add",
            target=str(group_id),
            payload_json=payload.dict(),
        )
    )
    await session.commit()
    return GroupMemberRead(account_name=member.account_name, connector_name=member.connector_name, weight=member.weight, enabled=member.enabled)


@router.patch(
    "/groups/{group_id}/members/{account_name}",
    response_model=GroupMemberRead,
    summary="更新分组成员",
    description="修改广播分组成员的权重（预留字段）或启用状态。",
)
async def update_group_member(
    group_id: int,
    account_name: str,
    payload: GroupMemberUpdateRequest,
    admin=Depends(require_admin),
    session: AsyncSession = Depends(get_session),
) -> GroupMemberRead:
    stmt = select(GroupMember).where(
        GroupMember.group_id == group_id,
        GroupMember.account_name == account_name,
    )
    result = await session.execute(stmt)
    member = result.scalar_one_or_none()
    if member is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="成员不存在")
    if payload.weight is not None:
        if payload.weight < 1:
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="权重必须为正整数")
        member.weight = payload.weight
    if payload.enabled is not None:
        member.enabled = payload.enabled
    if payload.connector_name is not None:
        connector = payload.connector_name.strip()
        group = await session.get(Group, group_id)
        if not group:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="分组不存在")
        if connector.lower() != group.connector_name.lower():
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="成员交易所需与分组一致")
        member.connector_name = connector
    await session.commit()
    await session.refresh(member)
    session.add(
        AuditLog(
            actor_id=admin.id,
            action="broadcast.group.member.update",
            target=str(group_id),
            payload_json=payload.dict(exclude_unset=True) | {"account_name": account_name},
        )
    )
    await session.commit()
    return GroupMemberRead(account_name=member.account_name, connector_name=member.connector_name, weight=member.weight, enabled=member.enabled)


@router.delete(
    "/groups/{group_id}/members/{account_name}",
    status_code=status.HTTP_204_NO_CONTENT,
    summary="移除分组成员",
    description="从广播分组中删除指定账户成员。",
)
async def remove_group_member(
    group_id: int,
    account_name: str,
    admin=Depends(require_admin),
    session: AsyncSession = Depends(get_session),
) -> Response:
    stmt = select(GroupMember).where(
        GroupMember.group_id == group_id,
        GroupMember.account_name == account_name,
    )
    result = await session.execute(stmt)
    member = result.scalar_one_or_none()
    if member is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="成员不存在")
    await session.delete(member)
    session.add(
        AuditLog(
            actor_id=admin.id,
            action="broadcast.group.member.delete",
            target=str(group_id),
            payload_json={"account_name": account_name},
        )
    )
    await session.commit()
    return Response(status_code=status.HTTP_204_NO_CONTENT)


@router.get(
    "/risk-config",
    summary="查看广播风控配置",
    description=(
        "读取当前广播相关的风控参数，包括交易所/交易对/订单类型白名单、默认名义金额上限、限流设置与滑点保护阈值。"
    ),
)
async def risk_config(_: object = Depends(require_admin)) -> dict:
    settings = get_settings()
    return {
        "allowed_connectors": settings.allowed_connectors_list,
        "allowed_symbols": settings.allowed_symbols_list,
        "allowed_order_types": settings.allowed_order_types_list,
        "max_notional_default": settings.max_notional_default,
        "webhook_rate_per_min": settings.webhook_rate_per_min,
        "user_rate_per_min": settings.user_rate_per_min,
        "slippage_guard": {"enabled": settings.slippage_guard_enabled, "bps_limit": settings.slippage_bps_limit},
    }


@router.post(
    "/broadcast",
    response_model=BroadcastResponse,
    status_code=status.HTTP_200_OK,
    summary="按分组广播下单",
    description=(
        "管理员按分组对多账户批量下单：\n"
        "- `group_id`：目标分组 ID；\n"
        "- `connector`：目标交易所连接器，需要在白名单内；\n"
        "- `trade_type` / `order_type` / `position_action`：BUY/SELL、MARKET/LIMIT、OPEN/CLOSE；\n"
        "- `symbol`：交易对（如 BTC-USDT）；\n"
        "- `price`：限价价格，可省略表示市价；\n"
        "- `invest_percent`：投入百分比（10 表示 10%），买单按报价资产权益计算名义金额，卖单按标的资产持仓计算数量并折算名义金额；\n"
        "- `skip_unbound`：是否跳过未绑定目标 connector 的账户（默认 false，拒单并提示）。\n"
        "接口会返回每个成员的下单结果与 hb-api 回执摘要。"
    ),
)
async def broadcast_order(
    payload: BroadcastOrderRequest,
    admin=Depends(require_admin),
    session: AsyncSession = Depends(get_session),
) -> BroadcastResponse:
    group = await session.get(Group, payload.group_id)
    if not group:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Group not found")
    if payload.connector.lower() != group.connector_name.lower():
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="请求的交易所需与分组绑定一致")

    stmt = select(GroupMember).where(GroupMember.group_id == payload.group_id, GroupMember.enabled.is_(True))
    result = await session.execute(stmt)
    members: List[GroupMember] = result.scalars().all()
    if not members:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="No active members in group")

    enforce_whitelists(payload.connector, payload.symbol, payload.order_type.value)
    try:
        rule = await hb_client.get_trading_rules(payload.connector, payload.symbol)
    except HBClientError as exc:
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail="无法连接 hummingbot-api，获取交易规则失败。",
        ) from exc

    settings = get_settings()
    results: List[BroadcastResultItem] = []
    order_book_cache: Dict[tuple[str, str], tuple[Optional[Decimal], Optional[Decimal]]] = {}
    trade_type_value = payload.trade_type.value
    order_type_value = payload.order_type.value
    position_action_value = payload.position_action.value
    market_type = _infer_market_type(payload.connector)

    connector_cache: Dict[str, Optional[set[str]]] = {}
    base_asset = _extract_base_asset(payload.symbol)
    quote_asset = _extract_quote_asset(payload.symbol)
    invest_ratio = payload.invest_percent / Decimal("100")
    is_buy = payload.trade_type == TradeSide.buy
    if settings.slippage_bps_limit and settings.slippage_bps_limit > 0:
        slippage_decimal = Decimal(str(settings.slippage_bps_limit)) / Decimal("10000")
    else:
        slippage_decimal = Decimal("0")

    for member in members:
        if member.connector_name.lower() == PLACEHOLDER_CONNECTOR:
            member.connector_name = payload.connector
        if member.connector_name.lower() != payload.connector.lower():
            results.append(
                BroadcastResultItem(
                    account_name=member.account_name,
                    status=OrderStatus.rejected,
                    error=_translate_error(code="member_connector_mismatch"),
                )
            )
            continue
        account_key = member.account_name.lower()
        if account_key not in connector_cache:
            try:
                connectors = await hb_client.list_account_connectors(member.account_name)
                connector_cache[account_key] = {c.lower() for c in connectors}
            except HBClientError as exc:
                logger.warning(
                    "Broadcast order failed to list connectors for %s: %s",
                    member.account_name,
                    exc,
                )
                connector_cache[account_key] = None

        connectors_set = connector_cache.get(account_key)
        if connectors_set is None:
            results.append(
                BroadcastResultItem(
                    account_name=member.account_name,
                    status=OrderStatus.error,
                    error=_translate_error(code="connector_lookup_error"),
                )
            )
            continue

        if payload.connector.lower() not in connectors_set:
            error_code = "skipped_missing_connector" if payload.skip_unbound else "missing_connector_binding"
            results.append(
                BroadcastResultItem(
                    account_name=member.account_name,
                    status=OrderStatus.rejected,
                    error=_translate_error(code=error_code),
                    tx_ref=None,
                )
            )
            if not payload.skip_unbound:
                session.add(
                    AuditLog(
                        actor_id=admin.id,
                        action="broadcast.reject.unbound",
                        target=str(payload.group_id),
                        payload_json={
                            "account_name": member.account_name,
                            "connector": payload.connector,
                        },
                    )
                )
            continue

        try:
            portfolio = await _fetch_member_portfolio(member.account_name, payload.connector)
        except HBClientError as exc:
            logger.warning("Failed to fetch portfolio for %s/%s: %s", member.account_name, payload.connector, exc)
            results.append(
                BroadcastResultItem(
                    account_name=member.account_name,
                    status=OrderStatus.rejected,
                    error=_translate_error(code="equity_fetch_failed"),
                )
            )
            continue

        quote_balance = _pick_balance_from_portfolio(portfolio, quote_asset)
        base_balance = _pick_balance_from_portfolio(portfolio, base_asset)

        explicit_price = Decimal(str(payload.price)) if payload.price is not None else None

        best_bid: Optional[Decimal] = None
        best_ask: Optional[Decimal] = None
        if order_type_value == OrderType.market.value:
            try:
                best_bid, best_ask = await signal_worker._get_best_quotes(
                    payload.connector,
                    payload.symbol,
                    order_book_cache,
                )
            except RiskError as exc:
                logger.warning(
                    "Broadcast order failed to fetch order book for %s/%s: %s",
                    payload.connector,
                    payload.symbol,
                    exc,
                )
                best_bid = best_ask = None

        notional = Decimal("0")
        if is_buy:
            if quote_balance <= 0:
                results.append(
                    BroadcastResultItem(
                        account_name=member.account_name,
                        status=OrderStatus.rejected,
                        error=_translate_error(code="insufficient_equity"),
                    )
                )
                continue
            notional = quote_balance * invest_ratio
        else:
            if base_balance <= 0:
                results.append(
                    BroadcastResultItem(
                        account_name=member.account_name,
                        status=OrderStatus.rejected,
                        error=_translate_error(code="insufficient_position"),
                    )
                )
                continue
            base_amount = base_balance * invest_ratio
            if base_amount <= 0:
                results.append(
                    BroadcastResultItem(
                        account_name=member.account_name,
                        status=OrderStatus.rejected,
                        error=_translate_error(code="insufficient_position"),
                    )
                )
                continue
            price_basis = explicit_price if explicit_price and explicit_price > 0 else None
            if price_basis is None:
                if order_type_value == OrderType.market.value:
                    price_basis = best_bid if best_bid and best_bid > 0 else None
                if price_basis is None or price_basis <= 0:
                    ref_price = get_reference_price(rule)
                    if ref_price > 0:
                        price_basis = ref_price
            if price_basis is None or price_basis <= 0:
                results.append(
                    BroadcastResultItem(
                        account_name=member.account_name,
                        status=OrderStatus.error,
                        error=_translate_error(code="price_unavailable"),
                    )
                )
                continue
            adjusted_price = price_basis
            if order_type_value == OrderType.market.value and slippage_decimal > 0:
                factor = Decimal("1") - slippage_decimal
                if factor <= 0:
                    results.append(
                        BroadcastResultItem(
                            account_name=member.account_name,
                            status=OrderStatus.error,
                            error=_translate_error(code="slippage_invalid"),
                        )
                    )
                    continue
                adjusted_price = price_basis * factor
            notional = base_amount * adjusted_price

        if notional <= 0:
            results.append(
                BroadcastResultItem(
                    account_name=member.account_name,
                    status=OrderStatus.rejected,
                    error=_translate_error(code="insufficient_equity" if is_buy else "insufficient_position"),
                )
            )
            continue

        try:
            enforce_max_notional(notional, None)
        except RiskError as exc:
            results.append(
                BroadcastResultItem(
                    account_name=member.account_name,
                    status=OrderStatus.rejected,
                    error=_translate_error(code=exc.code),
                )
            )
            continue

        try:
            normalized_qty, normalized_price = SignalWorker._compute_amount_and_price(
                market_type=market_type,
                order_type=order_type_value,
                trade_type=trade_type_value,
                position_action=position_action_value,
                notional=notional,
                explicit_price=explicit_price,
                rule=rule,
                best_bid=best_bid,
                best_ask=best_ask,
                slippage_bps=settings.slippage_bps_limit,
            )
        except RiskError as exc:
            results.append(
                BroadcastResultItem(
                    account_name=member.account_name,
                    status=OrderStatus.rejected,
                    error=_translate_error(code=exc.code),
                )
            )
            continue

        price_for_payload = (
            float(normalized_price)
            if normalized_price is not None
            else 0.0
        )

        order_payload = {
            "account_name": member.account_name,
            "connector_name": payload.connector,
            "trading_pair": payload.symbol,
            "trade_type": trade_type_value,
            "order_type": order_type_value,
            "position_action": position_action_value,
            "amount": float(normalized_qty),
            "price": price_for_payload,
        }

        raw_error: Optional[str] = None
        try:
            response = await hb_client.place_order(order_payload)
            remote_status = response.get("status", "").lower()
            if remote_status in {"filled", "completed"}:
                status_value = OrderStatus.filled
            elif remote_status in {"partial", "partially_filled"}:
                status_value = OrderStatus.partial
            else:
                status_value = OrderStatus.sent
            tx_ref = response.get("order_id") or response.get("id")
        except Exception as exc:  # noqa: BLE001
            status_value = OrderStatus.error
            tx_ref = None
            raw_error = str(exc)
            logger.error("Broadcast order failed for %s/%s: %s", member.account_name, payload.symbol, raw_error)
            response = {}

        error_text: Optional[str] = None
        if raw_error:
            error_text = _translate_error(raw=raw_error)
        else:
            response_error = response.get("error") if isinstance(response, dict) else None
            if response_error:
                error_text = _translate_error(raw=response_error)

        exec_order = ExecOrder(
            signal_id=None,
            user_id=None,
            account_name=member.account_name,
            connector=payload.connector,
            symbol=payload.symbol,
            side=payload.trade_type,
            order_type=payload.order_type,
            position_action=payload.position_action,
            qty=normalized_qty,
            price=normalized_price,
            status=status_value,
            error=error_text if status_value == OrderStatus.error else None,
            tx_ref=tx_ref,
        )
        session.add(exec_order)
        results.append(
            BroadcastResultItem(
                account_name=member.account_name,
                status=status_value,
                error=error_text if status_value != OrderStatus.filled else None,
                tx_ref=tx_ref,
            )
        )

    session.add(
        AuditLog(
            actor_id=admin.id,
            action="broadcast.execute",
            target=str(payload.group_id),
            payload_json={
                "group_id": payload.group_id,
                "connector": payload.connector,
                "trade_type": trade_type_value,
                "order_type": order_type_value,
                "position_action": position_action_value,
                "symbol": payload.symbol,
                "price": float(payload.price) if payload.price is not None else None,
                "invest_percent": float(payload.invest_percent),
            },
        )
    )
    await session.commit()
    return BroadcastResponse(results=results)


@public_router.get(
    "/groups",
    response_model=List[GroupSummary],
    summary="广播分组列表（普通用户）",
    description="普通用户查看管理员创建的广播分组，并标识自己是否已加入。",
)
async def list_public_groups(
    current_user=Depends(require_user),
    session: AsyncSession = Depends(get_session),
) -> List[GroupSummary]:
    stmt = (
        select(Group, func.count(GroupMember.id).label("member_count"))
        .outerjoin(GroupMember, Group.id == GroupMember.group_id)
        .group_by(Group.id)
        .order_by(Group.id.desc())
    )
    result = await session.execute(stmt)
    rows = result.all()

    account_name = default_account_name(current_user.id, current_user.email).lower()
    group_ids = [row[0].id for row in rows]
    joined_ids: set[int] = set()
    if group_ids:
        member_stmt = select(GroupMember.group_id).where(
            GroupMember.group_id.in_(group_ids),
            func.lower(GroupMember.account_name) == account_name,
        )
        member_result = await session.execute(member_stmt)
        joined_ids = {gid for (gid,) in member_result.all()}

    summaries: List[GroupSummary] = []
    for group, member_count in rows:
        summaries.append(
            GroupSummary(
                id=group.id,
                name=group.name,
                connector_name=group.connector_name,
                desc=group.desc,
                member_count=member_count,
                joined=group.id in joined_ids,
            )
        )
    return summaries


@public_router.post(
    "/groups/{group_id}/join",
    response_model=GroupMemberRead,
    status_code=status.HTTP_201_CREATED,
    summary="加入广播分组",
    description="普通用户加入指定广播分组，默认使用登录邮箱作为账户名，权重默认为 1（预留字段）。",
)
async def join_group(
    group_id: int,
    payload: GroupJoinRequest,
    current_user=Depends(require_user),
    session: AsyncSession = Depends(get_session),
) -> GroupMemberRead:
    group = await session.get(Group, group_id)
    if not group:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="分组不存在")
    if group.connector_name.lower() == PLACEHOLDER_CONNECTOR:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="分组尚未配置交易所，请联系管理员")

    account_name = payload.account_name.strip() if payload.account_name else default_account_name(current_user.id, current_user.email)
    if account_name.lower() != current_user.email.lower():
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="账户名必须等于登录邮箱")

    connector = payload.connector_name.strip()
    if connector.lower() != group.connector_name.lower():
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="选择的交易所需与分组一致")

    weight = payload.weight or 1
    if weight < 1:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="权重必须为正整数")

    try:
        connectors = await hb_client.list_account_connectors(account_name)
    except HBClientError as exc:
        logger.warning("Failed to list connectors for %s when joining group %s: %s", account_name, group_id, exc)
        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="查询账户交易所绑定失败，请稍后重试") from exc

    connector_set = {c.lower() for c in connectors}
    if connector.lower() not in connector_set:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="未绑定该交易所，请先完成交易所绑定")

    member = GroupMember(
        group_id=group_id,
        account_name=account_name,
        connector_name=connector,
        weight=weight,
        enabled=True,
    )
    session.add(member)
    try:
        await session.commit()
    except IntegrityError:
        await session.rollback()
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="已加入该分组")
    await session.refresh(member)

    session.add(
        AuditLog(
            actor_id=current_user.id,
            action="broadcast.group.join",
            target=str(group_id),
            payload_json={
                "account_name": account_name,
                "connector_name": connector,
                "weight": weight,
            },
        )
    )
    await session.commit()
    return GroupMemberRead(account_name=member.account_name, connector_name=member.connector_name, weight=member.weight, enabled=member.enabled)


@public_router.delete(
    "/groups/{group_id}/leave",
    status_code=status.HTTP_204_NO_CONTENT,
    summary="退出广播分组",
    description="普通用户退出指定广播分组（使用登录邮箱作为账户标识）。",
)
async def leave_group(
    group_id: int,
    current_user=Depends(require_user),
    session: AsyncSession = Depends(get_session),
) -> Response:
    account_name = default_account_name(current_user.id, current_user.email)
    stmt = select(GroupMember).where(
        GroupMember.group_id == group_id,
        func.lower(GroupMember.account_name) == account_name.lower(),
    )
    result = await session.execute(stmt)
    member = result.scalar_one_or_none()
    if member is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="未加入该分组")
    await session.delete(member)
    session.add(
        AuditLog(
            actor_id=current_user.id,
            action="broadcast.group.leave",
            target=str(group_id),
            payload_json={"account_name": account_name},
        )
    )
    await session.commit()
    return Response(status_code=status.HTTP_204_NO_CONTENT)
